/**
 * 
 */
package component.hadoop.statword;

import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;

/**
 * @author bliu
 * 
 */
public class WordCount {

	/**
	 * @param args
	 */
	public static void main(String[] args) throws Exception {

		if (args.length != 2) {
			System.err.println("Usage: wordcount <in> <out>");
			System.exit(2);
		}
		
		JobConf conf = new JobConf();
		// 如果Hadoop用户标识不同于客户机上的用户账号，可以通过hadoop.job.ugi属性来显式设定Hadoop用户名和组名。 
		// 用户名和组名之间有半角逗号隔开,其中第一个为用户名，后面则为组名。 
		//conf.set("Hadoop.job.ugi", "root,root123");

		Job job = new Job(conf);

		// 设置主函数入口
		job.setJarByClass(WordCount.class);
		// 设置输出的key
		job.setMapOutputKeyClass(Text.class);
		// 设置输出值
		job.setMapOutputValueClass(IntWritable.class);
		
		job.setMapperClass(StatMapper.class);
		// 合并函数
		// job.setCombinerClass(StatReduce.class);
		job.setReducerClass(StatReduce.class);
		
		job.setInputFormatClass(TextInputFormat.class);
		job.setOutputFormatClass(TextOutputFormat.class);

		FileInputFormat.setInputPaths(job, new Path(args[0]));
		FileOutputFormat.setOutputPath(job, new Path(args[1]));

		// 是否等待程序执行
		job.waitForCompletion(true);
	}

}
